////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
//                                                                                                                            //
//                                                         WNProject                                                          //
//                                                                                                                            //
//            This file is distributed under the BSD 2-Clause open source license. See License.txt for details.               //
//                                                                                                                            //
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

#include "WNNetworking/inc/WNNetworkManager.h"
#include "WNMemory/inc/WNMemory.h"
#include "WNCore/inc/WNAssert.h"
#include "WNMath/inc/WNBasic.h"
#include "WNContainers/inc/WNSerializer.h"
#include "WNConcurrency/inc/WNThread.h"
#include <sys/epoll.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/errno.h>
#include <fcntl.h>
#include <netdb.h>
#include <stdlib.h>
#include <algorithm>

using namespace WNNetworking;

WNNetworkManager::WNNetworkManager() :
    mMaxThreads(0),
    mInitializationState(WN_NOT_STARTED),
    mCurrentHandleVal(1){
}

WNNetworkManager::~WNNetworkManager() {
}

WNNetworkingError WNNetworkManager::Initialize(WN_UINT32 _numWorkerThreads) {
   
   for(WN_SIZE_T i = 0; i < _numWorkerThreads; ++i) {
        WNConcurrency::WNThread* thread = WN_NEW(WNConcurrency::WNThread(WNQueueThread, this));
        WN_RELEASE_ASSERT(thread != WN_NULL);
        mThreads.push_back(thread);
    }
    
    mInitializationState = WN_THREADS_CREATED;
   
    mReadThread = WN_NEW(WNConcurrency::WNThread(WNReadThread, this));
    WN_RELEASE_ASSERT(mReadThread != WN_NULL);
    mWriteThread = WN_NEW(WNConcurrency::WNThread(WNWriteThread, this));
    WN_RELEASE_ASSERT(mWriteThread != WN_NULL);

    mInitializationState = WN_LISTEN_THREAD_CREATED;

    mWriteEPollInstance = epoll_create1(0);
    if(mWriteEPollInstance == -1){
        Cleanup();
        return(WNNetworking::WN_NETWORKING_MANAGER_SYSTEM_INITIALIZATION_FAILED);
    }
    mInitializationState = WN_EPOLL_WRITE_CREATED;

    mReadEPollInstance = epoll_create1(0);
    if(mReadEPollInstance == -1) {
        Cleanup();
        return(WNNetworking::WN_NETWORKING_MANAGER_SYSTEM_INITIALIZATION_FAILED);
    }
    mInitializationState = WN_EPOLL_READ_CREATED;
    
    mInitializationState = WN_INITIALIZATION_COMPLETE;

    WN_RELEASE_ASSERT(mWriteThread->Execute());
    WN_RELEASE_ASSERT(mReadThread->Execute());
    for(WN_SIZE_T i = 0; i < mThreads.size(); ++i){
        WN_RELEASE_ASSERT(mThreads[i]->Execute());
    }
     
    return(WNNetworking::WN_OK);
}

WN_VOID WNNetworkManager::Cleanup() {
    switch(mInitializationState){
        case WN_INITIALIZATION_COMPLETE:
        case WN_EPOLL_READ_CREATED:
            close(mReadEPollInstance);
        case WN_EPOLL_WRITE_CREATED:
            close(mWriteEPollInstance);
        case WN_LISTEN_THREAD_CREATED:
            WN_DELETE(mReadThread);
            WN_DELETE(mWriteThread);
        case WN_THREADS_CREATED:
            for(WN_SIZE_T i = 0; i < mThreads.size(); ++i) {
                WN_DELETE(mThreads[i]);
            }
            mThreads.clear();
        default:
            break;
    }
    for(std::map<WNConnectionHandle, WNConnectionData*>::iterator i = mHandles.begin(); i != mHandles.end(); ++i){
        WN_DELETE(i->second);  
    }
    mHandles.clear();
    for(std::map<WNConnectionHandle, Group*>::iterator i = mGroups.begin(); i != mGroups.end(); ++i){
        WN_DELETE(i->second);
    }
    mGroups.clear();
}

WN_VOID WNNetworkManager::AddToReadEPoll(WNConnectionData* _data) {
    struct epoll_event event;
    WNMemory::WNMemSet(&event, 0, sizeof(epoll_event));
    event.data.ptr = _data;
    event.events = EPOLLIN | EPOLLONESHOT;
    if(epoll_ctl(mReadEPollInstance, EPOLL_CTL_MOD, _data->GetHandle(), &event) == -1) {
        // NOTIFY OF ERROR
        close(_data->GetHandle());
    }
}

WN_VOID WNNetworkManager::AddToWriteEPoll(WNConnectionData* _data) {
    struct epoll_event event;
    WNMemory::WNMemSet(&event, 0, sizeof(epoll_event));
    event.data.ptr = _data;
    event.events = EPOLLOUT | EPOLLONESHOT;
    if(epoll_ctl(mWriteEPollInstance, EPOLL_CTL_MOD, _data->GetHandle(), &event) == -1) {
        //NOTIFY ERROR
        close(_data->GetHandle());
    }
}

WNNetworkingError WNNetworkManager::CreateListener(WNConnectionHandle& _outHandle, WNConnectionType _type, WN_UINT16 _port, WNContainers::WNCallback2<WN_BOOL, WNConnectionHandle, const WN_CHAR*> _callback){
    WN_RELEASE_ASSERT_DESC(_type == eWNReliable, "WNNetworking does not support unreliable connections .... YET");
    struct addrinfo hints;
    struct addrinfo *result, *rp;
    WN_INT32 sock;
    WN_INT32 sockFD;

    WNMemory::WNMemSet(&hints, 0, sizeof(struct addrinfo));
    
    struct sockaddr_in serv_addr;
    WNMemory::WNMemSet(&serv_addr, 0, sizeof(serv_addr));
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    serv_addr.sin_port=htons(_port);

    sockFD = socket(AF_INET, SOCK_STREAM, 0);
    if(sockFD == -1) {
        return(WNNetworking::WN_NETWORKING_MANAGER_CANNOT_ASSOCIATE);
    }
    
    sock = bind(sockFD, (struct sockaddr*)&serv_addr, sizeof(serv_addr));
    if(sock != 0) {
        close(sockFD);
        return(WNNetworking::WN_NETWORKING_MANAGER_CANNOT_ASSOCIATE);
    }

    WN_INT32 flags;

    flags = fcntl(sockFD, F_GETFL, 0);
    if(flags == -1) {
        close(sockFD);
        return(WNNetworking::WN_NETWORKING_MANAGER_CANNOT_LISTEN);
    }
   
    flags |= SOCK_NONBLOCK;

    if(fcntl (sockFD, F_SETFL, flags) == -1) {
        close(sockFD);
        return(WNNetworking::WN_NETWORKING_MANAGER_CANNOT_LISTEN);
    }
   
    if(listen(sockFD, SOMAXCONN) == -1) {
        close(sockFD);
        return(WNNetworking::WN_NETWORKING_MANAGER_CANNOT_LISTEN);
    }

   
    WNConnectionHandle handleNum = 0;
    while((handleNum = WNAtomicCompareSwap(&mCurrentHandleVal, mCurrentHandleVal + 1, mCurrentHandleVal)) != mCurrentHandleVal - 1);
    WNConnectionHandle handleType = eWNSocketServer;
    handleType = handleType << CONNECTION_SHIFT;
    handleNum |= handleType;

    WNConnectionData* dat = WN_NEW(WNConnectionData(*this));
    
    dat->SetSocketDescriptor(sockFD);
    dat->SetConnectionHandle(handleNum);
    dat->SetConnectedCallback(_callback);

    mConnectionLock.Lock();
    mHandles[handleNum] = dat;
    mConnectionLock.Unlock();

    mListenLock.Lock();
    mListenSockets.push_back(dat);
    mListenLock.Unlock();

    struct epoll_event event;
    WNMemory::WNMemSet(&event, 0, sizeof(epoll_event));
    event.data.ptr = dat;
    event.events = EPOLLIN | EPOLLONESHOT;
    
    if(epoll_ctl(mReadEPollInstance, EPOLL_CTL_ADD, sockFD, &event) != 0) {
        close(sockFD);
        mConnectionLock.Lock();
        mHandles.erase(mHandles.find(handleNum));
        mConnectionLock.Unlock();

        mListenLock.Lock();
        mListenSockets.erase(std::find(mListenSockets.begin(), mListenSockets.end(), dat));
        mListenLock.Unlock();
        WN_DELETE(dat);
        return(WNNetworking::WN_NETWORKING_MANAGER_CANNOT_ASSOCIATE);
    }

    _outHandle = handleNum;
    return(WNNetworking::WN_OK);
}

::WNNetworkingError WNNetworkManager::ConnectTo(WNConnectionHandle& _outHandle, WNConnectionType _type, const WN_CHAR* _target, WN_UINT16 _port) {
    WN_RELEASE_ASSERT_DESC(_type == eWNReliable, "WNNetworking does not support unreliable connections .... YET");
    struct addrinfo* result;
    struct addrinfo* rp;
    struct addrinfo hints;
    
    WN_INT32 sock;
    WN_INT32 sockFD;

    WNMemory::WNMemSet(&hints, 0, sizeof(struct addrinfo));
    hints.ai_family = AF_INET;
    hints.ai_socktype = SOCK_STREAM;
    hints.ai_flags = AI_PASSIVE;
    WN_CHAR port[16];
    snprintf(port, 16, "%d", _port);
    sock = getaddrinfo(_target, port, &hints, &result);

    if(sock != 0) {
        return(WNNetworking::WN_NETWORKING_MANAGER_CANNOT_CREATE_SOCKET);
    }

    if(sockFD < 0){
        return(WNNetworking::WN_NETWORKING_MANAGER_CANNOT_CREATE_SOCKET);
    }

    for(rp = result; rp != NULL; rp = rp->ai_next) {
        sockFD = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
        if(sockFD == -1) {
            continue;
        }   
        connect(sockFD, rp->ai_addr, rp->ai_addrlen);
        if(sock == 0) {
            break;
        }
        close(sockFD);
    }

    if(rp == NULL) {
        return(WNNetworking::WN_NETWORKING_MANAGER_BAD_HOSTNAME);
    }

    freeaddrinfo(result);

    WN_INT32 flags;

    flags = fcntl(sockFD, F_GETFL, 0);
    if(flags == -1) {
        close(sockFD);
        return(WNNetworking::WN_NETWORKING_MANAGER_CANNOT_UNBLOCK);
    }
   
    flags |= SOCK_NONBLOCK;

    if(fcntl (sockFD, F_SETFL, flags) == -1) {
        close(sockFD);
        return(WNNetworking::WN_NETWORKING_MANAGER_CANNOT_UNBLOCK);
    }
   
   
    WNConnectionData* dat = WN_NEW(WNConnectionData(*this));
    
    WNConnectionHandle handleNum = 0;
    while((handleNum = WNAtomicCompareSwap(&mCurrentHandleVal, mCurrentHandleVal + 1, mCurrentHandleVal)) != mCurrentHandleVal - 1);
    WNConnectionHandle handleType = eWNSocketClient;
    handleType = handleType << CONNECTION_SHIFT;
    handleNum |= handleType;

    dat->SetSocketDescriptor(sockFD);
    dat->SetConnectionHandle(handleNum);

    mConnectionLock.Lock();
    mHandles[handleNum] = dat;
    mConnectionLock.Unlock();
    _outHandle = handleNum;

    struct epoll_event event;
    WNMemory::WNMemSet(&event, 0, sizeof(epoll_event));
    event.data.ptr = dat;
    event.events = EPOLLIN | EPOLLONESHOT;

    if(epoll_ctl(mReadEPollInstance, EPOLL_CTL_ADD, sockFD, &event) == -1) {
        WN_DELETE(dat);
        close(sockFD);
        mConnectionLock.Lock();
        mHandles.erase(mHandles.find(handleNum));
        mConnectionLock.Unlock();
        return(WNNetworking::WN_NETWORKING_MANAGER_CANNOT_ASSOCIATE);
    }
    WNMemory::WNMemSet(&event, 0, sizeof(epoll_event));
    event.events = EPOLLOUT | EPOLLONESHOT;
    event.data.ptr = dat;
    if(epoll_ctl(mWriteEPollInstance, EPOLL_CTL_ADD, sockFD, &event) == -1) {
        close(sockFD);
        return(WNNetworking::WN_NETWORKING_MANAGER_CANNOT_ASSOCIATE);
    }

    return(WNNetworking::WN_OK);
}

WN_VOID* WNNetworkManager::WNReadThread(WN_VOID* _networkManager) {
    WNNetworkManager* manager = reinterpret_cast<WNNetworkManager*>(_networkManager);
    WN_INT32 numEvents;
    static WN_INT32 MAX_EPOLL_EVENTS=64;
    static WN_INT32 MAX_EPOLL_TIME=5;
    epoll_event events[MAX_EPOLL_EVENTS];
    WNMemory::WNMemSet(events, 0, sizeof(epoll_event) * MAX_EPOLL_EVENTS);
   
    while(1){
        
        numEvents = epoll_wait(manager->mReadEPollInstance, events, MAX_EPOLL_EVENTS, -1);
        if(manager->mShuttingDown) {
            return(WN_NULL);
        }
        for(WN_INT32 i = 0; i < numEvents; ++i) {
            if((events[i].events & EPOLLERR)) {
                //ERROR;
                continue;
            } else if ((events[i].events & EPOLLHUP)) {
                //internal error?
                continue;
            } else if (!(events[i].events & EPOLLIN)) {
                //UHH whats this?
                continue;
            }
            manager->mListenLock.Lock();
            bool serverSocket = std::find (manager->mListenSockets.begin(), manager->mListenSockets.end(), events[i].data.ptr) != manager->mListenSockets.end();
            manager->mListenLock.Unlock();
            if(serverSocket) {
                WNConnectionMessage message;
                message.mType = eWNConnectionAcceptConnection;
                message.mData = reinterpret_cast<WNNetworkManager::WNConnectionData*>(events[i].data.ptr);
                manager->AddMessage(message);
            } else if(events[i].events & EPOLLIN) {
                WNConnectionMessage message;
                message.mType = eWNConnectionReadReady;
                message.mData = reinterpret_cast<WNNetworkManager::WNConnectionData*>(events[i].data.ptr);
                manager->AddMessage(message);
            } else {
                WN_DEBUG_ASSERT_DESC(WN_FALSE, "HOW DID WE GET HERE");
            }
        }
    }
}

WN_VOID* WNNetworkManager::WNWriteThread(WN_VOID* _networkManager) {
    WNNetworkManager* manager = reinterpret_cast<WNNetworkManager*>(_networkManager);
    WN_INT32 numEvents;
    static WN_INT32 MAX_EPOLL_EVENTS=64;
    static WN_INT32 MAX_EPOLL_TIME=5;
    epoll_event events[MAX_EPOLL_EVENTS];
   
    while(1){
        numEvents = epoll_wait(manager->mWriteEPollInstance, events, MAX_EPOLL_EVENTS, -1);
        if(manager->mShuttingDown) {
            WNConnectionMessage message;
            message.mType = eWNConnectionDone;
            for (WN_SIZE_T i = 0; i < manager->mThreads.size(); ++i) {
                manager->AddMessage(message);
            }
            for(WN_SIZE_T i = 0; i < manager->mThreads.size(); ++i) {
                manager->mThreads[i]->Join();
                WN_DELETE(manager->mThreads[i]);
            }
            manager->mThreads.clear();
            return(WN_NULL);
        }
        for(WN_INT32 i = 0; i < numEvents; ++i) {
            if((events[i].events & EPOLLERR) ||
               (events[i].events & EPOLLHUP) ||
               !((events[i].events & EPOLLOUT))) {
                //HANDLE CLOSING CONNECTION
                continue;
            }
            if(events[i].events & EPOLLOUT) {
                WNConnectionMessage message;
                message.mType = eWNConnectionSendReady;
                message.mData = reinterpret_cast<WNNetworkManager::WNConnectionData*>(events[i].data.ptr);
                manager->AddMessage(message);
            } else {
                WN_DEBUG_ASSERT_DESC(WN_FALSE, "HOW DID WE GET HERE");
            }
        }
    }
}

WN_VOID WNNetworkManager::AddMessage(const WNConnectionMessage& _message) {
    mQueueLock.Lock();
    mMessages.push_back(_message);
    mQueueLock.Unlock();
}

WN_VOID WNNetworkManager::AcceptConnection(WNConnectionData* _data) {
    while(1) {
        struct sockaddr addr;
        socklen_t len = sizeof(sockaddr);
        WN_INT32 sockFD;

        sockFD = accept(_data->GetHandle(), &addr, &len);
        if(sockFD == -1){
            if((errno == EAGAIN) || (errno == EWOULDBLOCK)) {
                break; //No more incomming connections now
            } else {
                char* s = strerror(errno);
                perror(NULL);
                //REPORT THE ERROR OR SOMETHING.. COULD NOT ACCEPT CONNECTION
                break;
            }
        }

        WN_CHAR name[NI_MAXHOST];
        WN_CHAR port[NI_MAXSERV];

        WN_BOOL validName = (getnameinfo(&addr, len, name, NI_MAXHOST, port, NI_MAXSERV, NI_NUMERICHOST | NI_NUMERICSERV) == 0);
        WN_INT32 flags;

        flags = fcntl(sockFD, F_GETFL, 0);
        if(flags == -1) {
            close(sockFD);
            //NOTIFY ERROR
            continue;
        }
   
        flags |= SOCK_NONBLOCK;

        if(fcntl (sockFD, F_SETFL, flags) == -1) {
            close(sockFD);
            //NOTIFY ERROR
            continue;
        } 
          
        WNConnectionData* dat = _data;

        WNConnectionHandle handleNum = 0;
        while((handleNum = WNAtomicCompareSwap(&mCurrentHandleVal, mCurrentHandleVal + 1, mCurrentHandleVal)) != mCurrentHandleVal - 1);
        WNConnectionHandle handleType = eWNSocketFromClient;
        handleType = handleType << CONNECTION_SHIFT;
        handleNum |= handleType;
 
        WNConnectionData* newdat = WN_NEW(WNConnectionData(*this));
        newdat->SetSocketDescriptor(sockFD);
        newdat->SetConnectionHandle(handleNum);
       
        mConnectionLock.Lock();
        mHandles[handleNum] = newdat;
        mConnectionLock.Unlock();

        struct epoll_event event;

        WNMemory::WNMemSet(&event, 0, sizeof(epoll_event));
        event.data.ptr = newdat;
        event.events = EPOLLIN | EPOLLONESHOT;
        if(epoll_ctl(mReadEPollInstance, EPOLL_CTL_ADD, sockFD, &event) == -1) {
            WN_DELETE(newdat);
            mHandles.erase(mHandles.find(handleNum));
            close(sockFD);
            //NOTIFY ERROR
            continue;
        }
        WNMemory::WNMemSet(&event, 0, sizeof(epoll_event));
        event.events = EPOLLOUT | EPOLLONESHOT;
        event.data.ptr = newdat;
        if(epoll_ctl(mWriteEPollInstance, EPOLL_CTL_ADD, sockFD, &event) == -1) {
            WN_DELETE(newdat);
            mHandles.erase(mHandles.find(handleNum));
            close(sockFD);
            //NOTIFY ERROR
            continue;
        }
        
        if(!dat->GetConnectionCallback().Execute(handleNum, validName? name : "\0")){
            WN_DELETE(newdat);
            mHandles.erase(mHandles.find(handleNum));
            close(sockFD);
            //NOTIFY ERROR
        }
    }
}

WN_VOID* WNNetworkManager::WNQueueThread(WN_VOID* _manager) {
    WNNetworkManager* manager = reinterpret_cast<WNNetworkManager*>(_manager);
    while(1){
        WNConnectionMessage m;
        while(true) {
            //DEAR GOD REPLACE THIS WHEN WE HAVE WORKING CONDITION VARS
            manager->mQueueLock.Lock();
            if(manager->mMessages.size() == 0){
                manager->mQueueLock.Unlock();
                WNConcurrency::WNThreadSleep(1);
                continue;
            }
            m = manager->mMessages.front();
            manager->mMessages.pop_front();
            manager->mQueueLock.Unlock();
            break;
        }
        
        switch(m.mType) {
            case eWNConnectionAcceptConnection:
                manager->AcceptConnection(m.mData);
                break;
            case eWNConnectionDone: {
                return(WN_NULL);
                break; 
            }
            case eWNConnectionSendReady: {
                m.mData->WriteReady(); 
                break;
            }
            case eWNConnectionReadReady: {
                m.mData->ReadReady(); 
                break;
            }
        }
        
    }
}

WN_VOID WNNetworkManager::WNConnectionData::ReadReady() {
   WN_SIZE_T processedBytes = 0;
   
   const WN_SIZE_T READ_BUFFER_SIZE = 1024 * 4;

   while(1){
        WNContainers::WNResourcePointer<WNBufferResource> p = mReadBuffers[0];     
        WN_UINT32 transferred = read(mSocketHandle, p->GetPointer(), p->GetSize() - p->GetWritten());
        if(transferred == -1){
            if(errno != EAGAIN) {
                char* s = strerror(errno);
                perror(NULL);
                //HANDLE ERROR HERE
                return;
            }
            break;
        } else if (transferred == 0) {
            //HANDLE CLOSED CONNECTION
            return;
        }
        WN_SIZE_T processedBytes = 0;
        mReadHead += transferred;
        mTotalReceived += transferred;
        
        while(processedBytes != transferred) {
            WN_DEBUG_ASSERT(mReadHead <= WNContainers::MAX_DATA_WRITE);
            if(mExpectedData == 0){
                if(mReadHead - mProcessedHead + mLeftoverData >= 8) {
                    WN_CHAR dataPlace[8];
                    for(WN_SIZE_T i = 0; i < mLeftoverData; ++i) {
                        dataPlace[i] = mLeftover[i];
                    }
                    for(WN_SIZE_T i = 0; i < (8 - mLeftoverData); ++i) {
                        dataPlace[mLeftoverData + i] = *(p->GetBaseLocation() + mProcessedHead + i);
                    }
                    WN_UINT32 num = *reinterpret_cast<WN_UINT32*>(dataPlace);
                    WN_UINT32 callback = *reinterpret_cast<WN_UINT32*>(dataPlace + 4);
                    WNFromBigEndian(callback);
                    WNFromBigEndian(num);
                    mCallbackFunction = callback;
                    WN_RELEASE_ASSERT(mCallbackFunction != 0);
                    mExpectedData = num;
                    WN_RELEASE_ASSERT(mExpectedData < 10);
                    mProcessedHead += (8 - mLeftoverData);
                    mProcessedData += (8 - mLeftoverData);
                    processedBytes += (8 - mLeftoverData);
                    mLeftoverData = 0;                 
                } else {
                    if (mReadHead - mProcessedHead + mLeftoverData > 8) {
                        WN_DEBUG_ASSERT(mProcessedHead == 0);
                        WN_CHAR* leftoverData = mCurrentReadBuffer.GetLastBuffer();
                        WN_CHAR dataPlace[8];
                        WN_SIZE_T ctr = 0;
                        for(ctr = 0; ctr < mLeftoverData; ctr++){
                            dataPlace[ctr] = leftoverData[WNContainers::MAX_DATA_WRITE - mLeftoverData + ctr];
                        }
                        for(WN_SIZE_T i = 0; i < (8 - ctr); ++i){
                            dataPlace[(8 - ctr) + i] = *(p->GetBaseLocation() + mProcessedHead + i);
                        }
                        WN_UINT32 num = *reinterpret_cast<WN_UINT32*>(dataPlace);
                        WNFromBigEndian(num);
                        mExpectedData = num;
                        mProcessedHead += 8 - mLeftoverData;
                        mProcessedData += 8 - mLeftoverData;
                        processedBytes += 8 - mLeftoverData;
                        mLeftoverData = 0;
                    } else {
                        //there are some leftover bytes
                        WN_SIZE_T  leftover = mReadHead - mProcessedHead;
                        for(WN_SIZE_T i = 0; i < leftover; ++i){
                            mLeftover[mLeftoverData + i] = *(p->GetBaseLocation() + mProcessedHead + i);
                        }
                        mLeftoverData += leftover;
                        mProcessedHead = mReadHead;
                        mProcessedData += leftover;

                        break;
                    }
                }
            }
            if(mExpectedData != 0) {
                WN_SIZE_T maxProcess = WNMath::WNMin(mReadHead - mProcessedHead, mExpectedData - mProcessedData);
                mProcessedData += maxProcess; //process as much as we can
                if(mProcessedData == mExpectedData) {
                    mCurrentReadBuffer.AppendBuffer(p, maxProcess, mProcessedHead);
                    WNContainers::WNCallback2<WN_VOID, WNConnectionHandle, WNNetworkReadBuffer&>* cb = mManager.GetCallback(mCallbackFunction);
                    WN_RELEASE_ASSERT_DESC(cb != WN_NULL, "Invalid callback for buffer");
                    cb->Execute(mConnectionHandle, mCurrentReadBuffer);
                    mCurrentReadBuffer.Clear();
                    mProcessedData = 0;
                    mExpectedData = 0;
                    mCallbackFunction = 0;
                }

                mProcessedHead += maxProcess;
                processedBytes += maxProcess;
            }
        }
        if(mReadHead == WNContainers::MAX_DATA_WRITE) {
            mReadBuffers[0] = WNContainers::WNAllocateResource<WNBufferResource>(mManager);
            mProcessedHead = mReadHead = 0;
        } else {
            mReadBuffers[0]->AddData(transferred);
        }
    }
   WN_RELEASE_ASSERT(mExpectedData < 10);
    mManager.AddToReadEPoll(this);    
}

WNNetworkingError WNNetworkManager::SendBuffer(WNConnectionHandle _handle, WNNetworkWriteBuffer& _buffer) {
    WN_SIZE_T buffInfo = (_handle >> CONNECTION_SHIFT) & CONNECTION_MASK;
    _buffer.FlushWrite();
    switch(buffInfo) {
        case eWNSocketServer:
            //HANDLE SENDING TO ALL CONNECTIONS FROM THIS SOCKET
            break;
        case eWNSocketFromClient:
        case eWNSocketClient: {
            mConnectionLock.Lock();
            WNConnectionData* handle = mHandles[_handle];
            mConnectionLock.Unlock();
            WN_DEBUG_ASSERT_DESC(handle, "Trying to send to a non-existant handle");
            handle->AppendSendBuffer(_buffer);
            break;
        }
        case eWNSocketGroup: {
            Group* g = mGroups[_handle];
            for(std::vector<WNConnectionHandle>::iterator i = g->mGroupedHandles.begin(); i != g->mGroupedHandles.end(); ++i){
                SendBuffer(*i, _buffer);
            }
            break;
        }
    }
    return(WN_OK);
}

WN_VOID WNNetworkManager::WNConnectionData::WriteReady() {
    mSendBufferLock.Lock();
    if(mSendBuffers.size() > 0) {
        Send();
    } else {
        mReadyToWrite = true;
    }
    mSendBufferLock.Unlock();
}

WN_VOID WNNetworkManager::WNConnectionData::AppendSendBuffer(WNNetworkWriteBuffer& _buff) {
    mSendBufferLock.Lock();
    mSendBuffers.push_back(_buff);
    if(mReadyToWrite) {
        Send();
    }
    mSendBufferLock.Unlock();
}

WN_VOID WNNetworkManager::WNConnectionData::Send() {
   while(1) {
        WN_SIZE_T count;
        WNNetworkWriteBuffer& buff = mSendBuffers[0];
        const WNNetworkWriteBuffer::WNBufferQueue& q = buff.GetChunks();
        WN_RELEASE_ASSERT(q.size() > 0);
        WN_SIZE_T buffLeft = q.size() - mBufferChunkCount;
        WN_SIZE_T writeLeft =  q[mBufferChunkCount]->GetWritten() - mBufferWritten;
        WN_RELEASE_ASSERT(writeLeft > 0);
        do{
            WN_INT32 s = write(mSocketHandle, q[mBufferChunkCount]->GetBaseLocation() + mBufferWritten, writeLeft);
            if(s > 0) {
                mBufferWritten += s;
                writeLeft -= s;
                mTotalSent += s;
            } else if(s == -1) {
                if(errno == EAGAIN) {
                    mReadyToWrite = false;
                    mManager.AddToWriteEPoll(this);    
                    return;
                }
                char* s = strerror(errno);
                perror(NULL);
                //ERROR CLOSE CONNNECTION; 
            } else { 
                char* s = strerror(errno);
                perror(NULL);
                //SOME REALLY BAD ERROR/
                WN_RELEASE_ASSERT_DESC(false, "WTF?");
            }
            if(writeLeft == 0) {
                buffLeft -= 1;
                if(buffLeft == 0) {
                    mSendBuffers.pop_front();
                    mBufferWritten = 0;
                    if(mSendBuffers.size() == 0){
                        mReadyToWrite = true;
                        return;
                    } 
                    break;
                } else {
                   mBufferChunkCount += 1;
                   mBufferWritten = 0;
                   writeLeft = q[mBufferChunkCount]->GetWritten(); 
                }
            }
       } while(true);
   }
}


WN_VOID WNNetworkManager::SetCallback(WN_UINT32 _identifier, WNContainers::WNCallback2<WN_VOID, WNConnectionHandle, WNNetworkReadBuffer&> _buffer) {
    WN_DEBUG_ASSERT(mCallbacks.find(_identifier) == mCallbacks.end());
    mCallbacks[_identifier] = _buffer;   
}

WNContainers::WNCallback2<WN_VOID, WNConnectionHandle, WNNetworkReadBuffer&>* WNNetworkManager::GetCallback(WN_UINT32 _identifier) {
    if(mCallbacks.find(_identifier) != mCallbacks.end()) {
        return(&mCallbacks[_identifier]);
    }
    return(WN_NULL);
}

WN_VOID WNNetworkManager::InitializeBuffer(WNNetworkWriteBuffer& _buffer, WN_UINT32 _number) {
    WN_UINT32 num = 0;
    _buffer.Serialize(0, WNContainers::WNSerializer<WN_UINT32>(num));
    _buffer.Serialize(0, WNContainers::WNSerializer<WN_UINT32>(_number));
}
